//*********************************************************
//
//    Copyright (c) Microsoft. All rights reserved.
//    This code is licensed under the Apache License, Version 2.0.
//    THIS CODE IS PROVIDED *AS IS* WITHOUT WARRANTY OF
//    ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING ANY
//    IMPLIED WARRANTIES OF FITNESS FOR A PARTICULAR
//    PURPOSE, MERCHANTABILITY, OR NON-INFRINGEMENT.
//
//*********************************************************

namespace WorkflowApplicationControls
{
    using System;
    using System.ComponentModel;
    using System.Threading;
    using Microsoft.Research.ScientificWorkflow.TridentUtilities;
    using Microsoft.Research.DataLayer;
    using System.Linq;
    using Microsoft.Research.eResearch.Common.Linq;

    /// <summary>
    /// This class monitors a job, and notifies on state change.
    /// </summary>
    public class JobMonitorHelper : IDisposable
    {
        #region Private Data

        private BackgroundWorker jobPollThread;

        private JobStatus currentJobStatus;

        private int outputCount;

        private Connection conn;

        #endregion Private Data

        #region Events

        /// <summary>
        /// Occurs when the state of a job changes.
        /// </summary>
        public event EventHandler<JobStatusEventArgs> OnJobChanged;

        #endregion Events

        #region Public Methods

        /// <summary>
        /// Starts the job monitor. It monitors the job status and outputs via a cloned connection of the passed connection.
        /// </summary>
        /// <param name="job">The job.</param>
        /// <param name="conn">The conn.</param>
        public void StartJobMonitor(IObject job, Connection connection)
        {
            if (null == job)
            {
                throw new ArgumentNullException("job");
            }

            if (null == connection)
            {
                throw new ArgumentNullException("connection");
            }

            // Clone the connection passed, else the data readers will throw exception due to multiple simaltaneous access.
            this.conn = null;
            try
            {
                this.conn = connection.Clone();
                this.conn.Open();

                // Open the job via the new connection.
                job = Job.Load((job as IObject).ID, this.conn);
            }
            catch (Exception ex)
            {
                TridentErrorHandler.HandleAndThrowBusinessLogicException(ex, TridentErrorConstants.ErrorNumber1300000006, WorkflowApplicationResourceManager.GetString("RegConnError"));
            }

            if (null == this.jobPollThread)
            {
                this.InitializeBackgroundWorker();
            }
            else
            {
                throw new InvalidOperationException(WorkflowApplicationResourceManager.GetString("BackgroundWorkerRunningError"));
            }

            // Reset the holder values and start the background worker.
            this.ResetStateObjects();
            this.jobPollThread.RunWorkerAsync(job);
        }

        #endregion Public Methods

        #region Private Methods

        private void InitializeBackgroundWorker()
        {
            this.jobPollThread = new BackgroundWorker();
            this.jobPollThread.WorkerReportsProgress = true;
            this.jobPollThread.WorkerSupportsCancellation = true;
            this.jobPollThread.DoWork += new DoWorkEventHandler(this.OnJobPoll);
            this.jobPollThread.ProgressChanged += new ProgressChangedEventHandler(this.OnJobProgressChanged);
            this.jobPollThread.RunWorkerCompleted += new RunWorkerCompletedEventHandler(this.OnJobPollCompleted);
        }

        private void OnJobPoll(object sender, DoWorkEventArgs e)
        {
            BackgroundWorker bgWorker = sender as BackgroundWorker;
            if (bgWorker != null && e != null && e.Argument != null)
            {
                try
                {
                    Job job = e.Argument as Job;
                    e.Result = this.MonitorJob(job, bgWorker, e);
                }
                catch (Exception ex)
                {
                    TridentErrorHandler.HandleAndThrowBusinessLogicException(ex, TridentErrorConstants.ErrorNumber1300000007, WorkflowApplicationResourceManager.GetString("OutputInfoRetrievalError"));
                }
            }
        }

        private void OnJobProgressChanged(object sender, ProgressChangedEventArgs e)
        {
            // On progress change, check for the user state passed, obtain the item changed and the changed job.
            // Based on what has been changed, invoke proper notifications.
            object[] userState = e.UserState as object[];
            if (userState != null && userState.Length > 1 && this.OnJobChanged != null)
            {
                JobItemChanged changedItem = (JobItemChanged)userState[1];
                Job modifiedJob = userState[0] as Job;
                if (modifiedJob != null && changedItem != JobItemChanged.None)
                {
                    this.OnJobChanged.Invoke(this, new JobStatusEventArgs() { ChangedJob = modifiedJob, ItemChanged = changedItem });
                }
            }
        }

        private void OnJobPollCompleted(object sender, RunWorkerCompletedEventArgs e)
        {
            this.ResetStateObjects();
            if (e.Error != null)
            {
                TridentErrorHandler.HandleAndThrowBusinessLogicException(e.Error, TridentErrorConstants.ErrorNumber1300000008, WorkflowApplicationResourceManager.GetString("OutputInfoRetrievalError"));
            }

            try
            {
                if (this.conn != null && this.conn.Alive)
                {
                    this.conn.Close();
                }
            }
            catch (Exception ex)
            {
                TridentErrorHandler.HandleUIException(ex);
            }
        }

        private Job MonitorJob(Job job, BackgroundWorker bgWorker, DoWorkEventArgs e)
        {
            if (null == job)
            {
                throw new ArgumentNullException("job");
            }

            int waitIndex = 0;

            // Monitor the job until its completed or aborted.
            JobItemChanged changedItem = JobItemChanged.None;
            while (!bgWorker.CancellationPending
                && job.Status != JobStatus.Aborted
                && job.Status != JobStatus.Completed)
            {
                try
                {
                    Thread.Sleep(Constants.REGISTRY_READ_WAIT_PERIOD);
                    changedItem = this.JobStateChanged(job);
                    Thread.Sleep(Constants.REGISTRY_READ_WAIT_PERIOD);
                    waitIndex++;
                }
                catch (Exception) { }

                if (waitIndex == Constants.EXECUTION_SERVICE_TIMEOUT_PERIOD && job.Status == JobStatus.Waiting)
                {
                    waitIndex = 0;
                    changedItem = JobItemChanged.Timeout;
                }

                if (changedItem != JobItemChanged.None)
                {
                    if (changedItem == JobItemChanged.FatalError)
                    {
                        bgWorker.CancelAsync();
                    }

                    bgWorker.ReportProgress(0, new object[] { job, changedItem });
                }
            }
            
            if (bgWorker.CancellationPending)
            {
                e.Cancel = true;
            }

            return job;
        }

        private JobItemChanged JobStateChanged(Job job)
        {
            // Check for the status changed and the outputs changed.
            // Return appropriate change indication.
            JobItemChanged stateChanged = JobItemChanged.None;
            if (this.conn.Alive)
            {
                job.Refresh();
                if (this.currentJobStatus != job.Status)
                {
                    this.currentJobStatus = job.Status;
                    stateChanged = JobItemChanged.Status;
                }

                try
                {
                    job.ActivityInstances.GetEnumerator();
                    if (job.ActivityInstances.Count > 0)
                    {
                        job.ActivityInstances[0].Outputs.GetEnumerator();
                        if (job.ActivityInstances[0].Outputs.Count > 0 && this.outputCount != job.ActivityInstances[0].Outputs.Count)
                        {
                            this.outputCount = job.ActivityInstances[0].Outputs.Count;
                            stateChanged = (stateChanged == JobItemChanged.Status) ? JobItemChanged.StatusAndOutputs : JobItemChanged.Outputs;
                        }
                    }
                }
                catch (Exception) { }
            }
            else if (job.Status != JobStatus.Aborted && job.Status != JobStatus.Completed)
            {
                // When the connection is not alive, the job is considered to be aborted, explicitly set the 
                // status on the job because if the conn is dead, it will never be updated.
                stateChanged = JobItemChanged.FatalError;
            }

            return stateChanged;
        }

        private void ResetStateObjects()
        {
            this.currentJobStatus = JobStatus.NotStarted;
            this.outputCount = 0;
        }

        #endregion Private Methods
        
        #region IDisposable Members

        /// <summary>
        /// Operation to perform while Disposing the object.
        /// </summary>
        public void Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }

        /// <summary>
        /// nullify and Disposes the object 
        /// </summary>
        /// <param name="disposing"></param>
        private void Dispose(bool disposing)
        {
            if (disposing && this.jobPollThread != null)
            {
                this.jobPollThread.Dispose();
            }
        }

        #endregion
    }

    /// <summary>
    /// Identifies the changed object.
    /// </summary>
    public enum JobItemChanged
    {
        /// <summary>
        /// Nothing has changed.
        /// </summary>
        None = 0,

        /// <summary>
        /// The job status has changed.
        /// </summary>
        Status,

        /// <summary>
        /// The output has changed.
        /// </summary>
        Outputs,

        /// <summary>
        /// Both the status and outputs have changed.
        /// </summary>
        StatusAndOutputs,

        /// <summary>
        /// Timeout
        /// </summary>
        Timeout,

        /// <summary>
        /// Indicates a fatal error, for e.g., the sql service being stopped.
        /// </summary>
        FatalError
    }

    /// <summary>
    /// The event args for the event invoked when a job is changed.
    /// </summary>
    public class JobStatusEventArgs : EventArgs
    {
        /// <summary>
        /// Gets or sets the item changed.
        /// </summary>
        /// <value>The item changed.</value>
        public JobItemChanged ItemChanged { get; set; }

        /// <summary>
        /// Gets or sets the changed job.
        /// </summary>
        /// <value>The changed job.</value>
        public Job ChangedJob { get; set; }
    }
}
